package worker

import (
	"context"
	"time"

	"gitee.com/taotao_today/crontab/crontab/common"
	"go.etcd.io/etcd/api/v3/mvccpb"

	clientv3 "go.etcd.io/etcd/client/v3"
)

// JobMgr 任务管理器
type JobMgr struct {
	client  *clientv3.Client
	kv      clientv3.KV
	lease   clientv3.Lease
	watcher clientv3.Watcher
}

var G_jobMgr *JobMgr

// InitJobMgr 连接ETCD
func InitJobMgr() (err error) {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   G_config.EtcdEndPoints,
		DialTimeout: time.Duration(G_config.EtcdDialTimeout) * time.Millisecond,
	})
	if err != nil {
		return
	}

	G_jobMgr = &JobMgr{
		client:  cli,
		kv:      clientv3.NewKV(cli),
		lease:   clientv3.NewLease(cli),
		watcher: clientv3.NewWatcher(cli),
	}

	// 启动任务监听
	err = G_jobMgr.WatchJobs()

	G_jobMgr.WatchKiller()
	return
}

func (m *JobMgr) CreateJobLock(jobName string) *JobLock {
	return InitJobLock(jobName, m.kv, m.lease)
}

// WatchJobs 监听任务变化
func (m *JobMgr) WatchJobs() (err error) {
	// 1. 获取当前/cron/jobs/目录下的所有job，并获知当前集群的version(用于后续监听)
	getResp, err := m.client.Get(context.TODO(), common.JOB_SAVE_DIR, clientv3.WithPrefix())
	if err != nil {
		return
	}
	for _, kv := range getResp.Kvs {
		job, err := common.UnpackJob(kv.Value)
		if err == nil {
			// todo: 将这个job同步给schedule(调度协程)
			jobEvent := common.BuildJobEvent(common.JOB_EVENT_SAVE, job)
			G_scheduler.PushJobEvent(jobEvent)
		}
	}

	// 2. 从该revision向后监听变化时间
	go func() {
		var jobEvent *common.JobEvent
		// 从GET时刻的后续版本开始监听变化
		watchStartRevision := getResp.Header.Revision + 1
		// 监听/cron/jobs/目录的后续变化
		watchChan := m.watcher.Watch(context.TODO(), common.JOB_SAVE_DIR, clientv3.WithRev(watchStartRevision), clientv3.WithPrefix())
		// 处理监听事件
		for watchResp := range watchChan {
			for _, watchEvent := range watchResp.Events {
				switch watchEvent.Type {
				case mvccpb.PUT: // 任务保存事件
					// 反序列化Job，推送给调度协程
					job, err := common.UnpackJob(watchEvent.Kv.Value)
					if err != nil {
						// 非法字符串
						continue
					}
					// 构建一个更新Event
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, job)
				case mvccpb.DELETE: // 任务被删除了
					// 推送一个删除事件给调度协程
					// Delete /cron/jobs/job1
					jobName := common.ExtractJobName(string(watchEvent.Kv.Key))

					job := &common.Job{Name: jobName}

					// 构建一个删除Event
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_DELETE, job)
				}
				// 变化推给scheduler
				G_scheduler.PushJobEvent(jobEvent)
			}

		}
	}()
	return
}

// WatchKiller 监听强杀任务通知
func (m *JobMgr) WatchKiller() {
	go func() {
		var jobEvent *common.JobEvent
		// 监听/cron/killer/目录的后续变化
		watchChan := m.watcher.Watch(context.TODO(), common.JOB_KILLER_DIR, clientv3.WithPrefix())
		// 处理监听事件
		for watchResp := range watchChan {
			for _, watchEvent := range watchResp.Events {
				switch watchEvent.Type {
				case mvccpb.PUT: // 任务保存事件
					jobName := common.ExtractKillerName(string(watchEvent.Kv.Key))
					job := &common.Job{Name: jobName}
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_KILL, job)
				case mvccpb.DELETE: // 任务被删除了
				}
				// 变化推给scheduler
				G_scheduler.PushJobEvent(jobEvent)
			}
		}
	}()
	return
}
